package com.strato.rpc.hessian.v1_0_0.ha;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import com.strato.base.cache.v1_0_0.TimedReference;
import com.strato.base.env.v1_0_0.GlobalEnvironment;
import com.strato.logging.v1_0_0.api.Log;
import com.strato.logging.v1_0_0.api.LogFactory;
import com.strato.rpc.hessian.v1_0_0.HAStrategy;
import com.strato.rpc.hessian.v1_0_0.HessianHttpTransport;

public abstract class AbstractHAStrategy implements HAStrategy {

	protected Log logger=LogFactory.getLog(getClass());

	protected String path;

	protected Class<?> intf;

	protected HessianHttpTransport transport;

	public AbstractHAStrategy(String path, Class<?> intf,HessianHttpTransport transport) {
		super();
		this.path = path;
		this.intf = intf;
		this.transport=transport;
		getProxies();
	}

	@Override
	public Object getProxy(String id) {
		return getProxies().get(id);
	}

	protected synchronized InvocationStatics getStat(String path) {
		InvocationStatics stat = stats.get(path);
		if(stat==null){
			logger.info("Found rpc target "+path);
			stat=new InvocationStatics();
			stats.put(path, stat);
		}
		return stat;
	}
	
	@Override
	public void markUnavailable(String path) {
		InvocationStatics stat=getStat(path);
		stat.markUnavailable();
	}

	@Override
	public void increaseCount(String path, int i) {
		InvocationStatics stat = stats.get(path);
		stat.getCount().incrementAndGet();
	}

	@Override
	public void increaseCost(String path, long l) {
		InvocationStatics stat = stats.get(path);
		stat.getCost().addAndGet(l);
	}

	protected TimedReference<Map<String,Object>> proxiesRef;
	protected synchronized Map<String,Object> getProxies(){
		if(proxiesRef==null||proxiesRef.isExpired()||proxiesRef.getObject()==null){
			String key="${strato.rpc.hessian.baseurls}";
			String value=GlobalEnvironment.getInstance().resolveString(key);
			if(key.equals(value)){
				throw new IllegalArgumentException("No global environment property "+key+" found");
			}
			Map<String,Object> proxies=new HashMap<String,Object>();
			String[] baseurls=value.split(",");
			for(String baseurl:baseurls){
				Object proxy = transport.createSingleProxy(baseurl,path,intf);
				proxies.put(baseurl, proxy);
				getStat(baseurl);
			}
			int cacheTimeout=60000;
			proxiesRef=new TimedReference<Map<String,Object>>(proxies,cacheTimeout);
		}
		return proxiesRef.getObject();
	}
	
	protected Map<String,InvocationStatics> stats=new HashMap<String,InvocationStatics>();
	protected class InvocationStatics{
		private AtomicInteger count=new AtomicInteger(0);
		private AtomicLong cost=new AtomicLong(0);
		private AtomicBoolean available=new AtomicBoolean(false);
		private Long lastFall=null;
		public AtomicInteger getCount() {
			return count;
		}
		public AtomicLong getCost() {
			return cost;
		}
		public void markUnavailable(){
			available.set(false);
			lastFall=System.currentTimeMillis();
		}
		public AtomicBoolean getAvailable() {
			return available;
		}
		public boolean isAvailable() {
			return available.get();
		}
		public Long getLastFall() {
			return lastFall;
		}
		public void markAvailable() {
			available.set(true);
		}
	}
}
